package com.thp.bigdata.webLogWash;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/**
 * 处理的主类
 * @author 汤小萌
 *
 */
public class WebLogPreProcess {
	static class WebLogPreProcessMapper extends Mapper<LongWritable, Text, Text, NullWritable> {
		Text k = new Text();
		@Override
		protected void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			String line = value.toString();
			// System.out.println(line);
			WebLogBean webLogBean = WebLogParser.parser(line);
			// System.out.println(webLogBean.isValid() ? "有效" : "无效");
			if(!webLogBean.isValid()) {
				return;   // 无效的数据
			}
			k.set(webLogBean.toString());
			context.write(k, NullWritable.get());
		}
	}
	
	public static void main(String[] args) throws IllegalArgumentException, IOException, ClassNotFoundException, InterruptedException {
		Configuration conf = new Configuration();
		Job job = Job.getInstance(conf);

		job.setJarByClass(WebLogPreProcess.class);

		job.setMapperClass(WebLogPreProcessMapper.class);

		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(NullWritable.class);

		FileInputFormat.setInputPaths(job, new Path("f:/webLog/input"));
		FileOutputFormat.setOutputPath(job, new Path("f:/webLog/output"));

		System.exit(job.waitForCompletion(true) ? 0 : 1);
	}
}
